#include <sys/types.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "log.h"
#include "queue.h"
#include "rqueue.h"

static QueueErr open_next_file(int index);
static QueueErr open_and_seek (int index);

static inline int position_set(int position, queue_status *qs)
{
	qs->position = position;
	return msync(qs, sizeof(queue_status), MS_SYNC);
}

static QueueErr open_next_file(int index)
{
	logger(PRIORITY_DEBUG, "NO.%d start switch to next queue file, nowid:%d", index, status[index]->id_now);
	if(status[index]->id_next == -1)
	{
		/* no next file found. */
		logger(PRIORITY_DEBUG, "no next file found");
		return QERR_OPEN_FILE;
	}

	status[index]->id_now = status[index]->id_next;
	if(status[index]->id_next == 0)
		/* no next file at all, we get the last queue file. */
		status[index]->id_next = -1;

	else
	{
		char filename[FILE_PATH_SIZE];
		/* check if the file next of the 'next file' is exist. */
		sprintf(filename, "%s.%d", handler[index].name, ++status[index]->id_next);
		if(-1 == access(filename, F_OK))
			/* now id_next come to the last queue file . */
			status[index]->id_next = 0;
	}
	
	if(-1 == msync(status[index], sizeof(queue_status), MS_SYNC))
		return QERR_OPEN_FILE;
	
	if(-1 == position_set(0, status[index]))
		return QERR_MSYNC;
	
	logger(PRIORITY_DEBUG, "NO.%d switch to next queue file success", index);
	return open_and_seek(index);
}

static QueueErr open_and_seek(int index)
{
	int position = 0;
	const char *name = handler[index].name;
	if(status[index]->id_now == 0) {
		if(-1 == (handler[index].rfd = open(name, O_RDWR|O_CREAT, 0644)))
			return QERR_OPEN_FILE;
	}
	else
	{
		char file[FILE_PATH_SIZE];
		sprintf(file, "%s.%d", name, status[index]->id_now);
		if(-1 == (handler[index].rfd = open(file, O_RDONLY, 0644)))
			return QERR_OPEN_FILE;
	}
	
	position = status[index]->position;
	if(0 < position && -1 == lseek(handler[index].rfd, position, SEEK_SET))
	{
		close(handler[index].rfd);
		handler[index].rfd = 0;
		return QERR_SEEK_FILE;
	}

	logger(PRIORITY_DEBUG, "NO.%d open queue file success, nowid:%d position:%d.",
		index, status[index]->id_now, position);
	return Q_EXEC_OK;
}

/* return value: 1 success; 0 end of queue; -1 some error occur.
 */
int next(int index, queue_header *header, void **value)
{
	char *ptr;
	int nbyte, fd, max_retry = 0;
	
start_get:
	fd = handler[index].rfd;
	if(status[index]->is_lock)
	{
		/* if the queue is locked by wqueue */
		close(fd);
		handler[index].rfd = 0;
		
		while(max_retry++ < 10)
		{
			// waiting for wqueue
			if(!status[index]->is_lock)
				break;
			usleep(300000);
		}
		
		if(Q_EXEC_OK != open_and_seek(index))
			goto read_data_error;

		fd = handler[index].rfd;
	}
	
	if(-1 == (nbyte = read(fd, header, sizeof(queue_header))))
		goto read_data_error;
	
	if(0 == nbyte)
	{
		/* at the end of file. */

		if(-1 == status[index]->id_next)
		{
			/* at the end of queue. */
#ifdef DEBUG	
			fprintf(stderr, "At end of queue(%d).\n", index);
			logger(PRIORITY_DEBUG, "At end of queue(%d) at position: %d.", index, status[index]->position);
#endif
			return 0;
		}

		close(fd);
		if(Q_EXEC_OK != open_next_file(index))
		{
#ifdef DEBUG
			fprintf(stderr, "Can't open next file at %d.\n", status[index]->id_now);
#endif
			logger(PRIORITY_ERROR, "switch to next file failure. at %d, at %d in '%s'.",
				status[index]->id_now, __LINE__, __FILE__);
			goto read_data_error;
		}
		
		goto start_get;
	}
	
	if(sizeof(queue_header) != nbyte || -1 != header->start_sign || !header->size)
	{
		// Error data in queue file.
#ifdef DEBUG
		fprintf(stderr, "Error data length, nbtye: %d.\n", nbyte);
#endif
		logger(PRIORITY_ERROR, "Error data length, nbtye: %d, at %d in '%s'.",
			nbyte, __LINE__, __FILE__);	
		lseek(fd, -nbyte, SEEK_CUR);
		goto read_data_error;
	}
	
	ptr = *value = (char*)malloc(header->size + 1);
	if(-1 == (nbyte = read(fd, ptr, header->size)))
	{
		free(*value);
		*value = NULL;
		goto read_data_error;
	}
	
	if(nbyte != header->size)
	{
		// Error data in queue file.
		free(*value);
		*value = NULL;
#ifdef DEBUG
		fprintf(stderr, "Error data in queue.\n");
#endif
		logger(PRIORITY_ERROR, "Error data in queue at %d in '%s'.",
				__LINE__, __FILE__);	
		lseek(fd, -nbyte-sizeof(queue_header), SEEK_CUR);
		goto read_data_error;
	}
	
	ptr[header->size] = '\0';
	return 1;
	
read_data_error:
#ifdef DEBUG
	fprintf(stderr, "Read data Error at [%d, %d].\n", status[index]->id_now , status[index]->position);
#endif
	logger(PRIORITY_ERROR, "Read data Error at [%d, %d], at %d in '%s'.",
			status[index]->id_now , status[index]->position, __LINE__, __FILE__);
	return -1;	
}

int forward(int index)
{
	int pos = 0;
	if(-1 != (pos = lseek(handler[index].rfd, 0, SEEK_CUR)))
	{
		if(-1 == position_set(pos, status[index])) {
#ifdef DEBUG
			fprintf(stderr, "Set position Error.");
#endif
			logger(PRIORITY_ERROR, "Set position Error at %d in '%s'.", __LINE__, __FILE__);
			return -1;
		}
	}
	
	return pos;	
}

int next_forward(int index, queue_header *header, void** value)
{
	int pos = 0;
	if(-1 == next(index, header, value)) return -1;
	if(-1 != (pos = lseek(handler[index].rfd, 0, SEEK_CUR)))
	{
		if(-1 == position_set(pos, status[index]))
		{
			// Error data in queue file.
			free(*value);
			lseek(handler[index].rfd, -header->size-sizeof(queue_header), SEEK_CUR);
#ifdef DEBUG
			fprintf(stderr, "Set position Error.");
#endif
			logger(PRIORITY_ERROR, "Set position Error at %d in '%s'.", __LINE__, __FILE__);
			return -1;
		}
	}
	
	return pos;
}

QueueErr rqueue_open(int index)
{
	if(Q_EXEC_OK != open_and_seek(index))
	{
#ifdef DEBUG
		fprintf(stderr, "Load queue failue.\n");
#endif
		logger(PRIORITY_ERROR, "Load queue failue at %d in '%s'.",
			__LINE__, __FILE__);
		
		return QERR_HARD_DISK;
	}
	
	return Q_EXEC_OK;
}

int rqueue_close(index)
{
	int fd = handler[index].rfd;
	return (-1 != close(fd)) & stat_close(status[index]);
}
